package com.example.storm.storm;

import com.alibaba.fastjson.JSON;
import com.example.storm.bean.InfoGet;
import com.example.storm.constant.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Map;

@Slf4j
public class Bolt extends BaseRichBolt{

	private static final long serialVersionUID = 6542256546124282695L;

	@Override
	public void prepare(Map map, TopologyContext arg1, OutputCollector collector) {
		//可以用来在这里给对象赋值
	}

	@Override
	public void execute(Tuple tuple) {
		String msg = tuple.getStringByField(Constants.FIELD);
		try{
			List<InfoGet> list = JSON.parseArray(msg, InfoGet.class);
			if(CollectionUtils.isEmpty(list)){
				return;
			}

			list.forEach(info->{
				if(info.getAge()>10){
					log.info("Bolt.execute 数据经过处理合格，输出，此条记录为:{}", info.toString());
				}
			});
		}catch(Exception e){
			log.error("Bolt的数据处理失败!数据:{}", msg, e);
		}
	}

	/**
	 * cleanup是IBolt接口中定义,用于释放bolt占用的资源。
	 * Storm在终止一个bolt之前会调用这个方法。
	 */
	@Override
	public void cleanup() {
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {

	}
}
